package com.example.flinkcourse.lesson1.transform;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 自定义 FlatMap 转换函数
 * 用于嵌套结构扁平化
 */
public class MyFlatMapFunction implements FlatMapFunction<String, String> {
    private static final Logger LOG = LoggerFactory.getLogger(MyFlatMapFunction.class);

    @Override
    public void flatMap(String value, Collector<String> out) throws Exception {
        LOG.debug("Processing record: {}", value);
        
        // 示例：将输入字符串按空格分割，并输出每个单词
        String[] words = value.split("\\s+");
        for (String word : words) {
            if (!word.isEmpty()) {
                LOG.debug("Emitting word: {}", word);
                out.collect(word);
            }
        }
    }
} 